#include <assert.h>
#include "socket_impl.h"
#include "socket_inc.h"
#include <core/QbLogger.h>
#include <core/system/sleep.h>
#include <core/system/guard.h>
#include <qbprotocol/include/ssCmnDefine.h>

namespace qb
{
    CSocketImpl::CSocketImpl()
        : m_port(0)
        , m_socket_type(IPPROTO_TCP)
        , m_observer(NULL)
        , m_critSect(*CriticalSectionWrapper::CreateCriticalSection())
        , m_crit_interface(*CriticalSectionWrapper::CreateCriticalSection())
        , m_event_send(*EventWrapper::Create(false))
        , m_thread_send(NULL)
        , m_exit_flag_send(0)
        , m_thread_recv(NULL)
        , m_exit_flag_recv(0)
        , m_udp_associate(NULL)
        , m_udp_proxy(false)
    {

    }

    CSocketImpl::~CSocketImpl()
    {
        delete &m_critSect;
        delete &m_crit_interface;
        delete &m_event_send;
        if (m_udp_associate)
        {
            delete m_udp_associate;
            m_udp_associate = NULL;
        }
    }

    int32_t CSocketImpl::RegisterSocketObserver(ISocketObserver* observer)
    {
        CriticalSectionScoped cs(&m_critSect);
        if (m_observer)
        {
            log_error("already enabled");
            return -1;
        }

        m_observer = observer;

        return 0;
    }

    int32_t CSocketImpl::DeRegisterSocketObserver()
    {
        CriticalSectionScoped cs(&m_critSect);
        if (!m_observer)
        {
            log_error("already disabled");
            return -1;
        }

        m_observer = NULL;

        return 0;
    }

    uint32_t CSocketImpl::GetTargetIP()
    {
        return m_socket.GetTargetIP();
    }

    uint16_t CSocketImpl::GetTargetPort()
    {
        return m_socket.GetTargetPort();
    }

    int32_t CSocketImpl::GetSocket()
    {
        return (int32_t)m_socket.GetSocket();
    }

    void CSocketImpl::SetSocks5UdpProxy(bool use_udp)
    {
        m_udp_proxy = use_udp;
    }

    int32_t CSocketImpl::Connect(const char* ip_address,uint16_t port,uint16_t socket_type)
    {
        {
            CriticalSectionScoped cs(&m_crit_interface);
            if (!m_udp_associate)
            {
                m_udp_associate = new CSocketImpl();
                m_udp_associate->SetSocks5UdpProxy(true);
            }

            if (m_socket.GetSocket() != INVALID_SOCKET)
            {
                log_error("socket be used");
            //    return -1;
            }

            if (socket_type != IPPROTO_TCP && socket_type != IPPROTO_UDP)
            {
                log_error("%socket type failed(%d)",socket_type);
                return -1;
            }
            m_socket_type	= socket_type;
            m_ipaddress		= ip_address;
            m_port			= port;

            // ??????
            // +------------------+-------+------+
            // | ???????		  | TCP	  | UDP  |
            // +------------------+-------+------+
            // | PROXY_BROWSER	  | ??	  | ??   |
            // +------------------+-------+------+
            // | PROXY_HTTP		  | ??	  | ??   |
            // +------------------+-------+------+
            // | PROXY_SOCKS_VER5 | ??	  | ??   |
            // +------------------+-------+------+
            if (m_proxy.GetProxyType() != kProxyNone && socket_type == IPPROTO_UDP)
            {
                log_info("UDP does not support proxy, automatically change to TCP");
                m_socket_type = IPPROTO_TCP;
            }

            if (m_socket.Create(socket_type) == -1)
            {
                log_error("socket failed(err=%d)",CSocketWrapper::GetSockError());
                return -1;
            }

            if (m_socket_type == IPPROTO_TCP)
            {
                if (m_socket.SetOpt() == -1)
                {
                    log_warning("%SetOpt failed");
                }
                if (m_socket.SetTcpOpt() == -1)
                {
                    log_warning("SetTcpOpt failed");
                }
                if (m_proxy.GetProxyType() != kProxyNone)
                {
                    if (m_socket.Connect(m_proxy.GetIPAddress(),m_proxy.GetPort()) == -1
                        || m_proxy.Connect(m_socket.GetSocket(),ip_address,port,m_udp_proxy) == -1)
                    {
                        log_error("tcp proxy connect failed");
                        m_socket.Close();
                        return -1;
                    }
                }
                else
                {
                    if (m_socket.Connect(ip_address,port) == -1)
                    {
                        log_error("tcp connect failed");
                        m_socket.Close();
                        return -1;
                    }
                }
                m_socket.SetKeepalive();

                m_event_send.Reset();
                ++m_exit_flag_send;
                uint32_t threadId = 0;
                m_thread_send = ThreadWrapper::CreateThread(ThreadFunSend,this,kNormalPriority,"socket send thread");
                m_thread_send->Start(threadId);
            }
            else
            {
                if (m_socket.GetIPAddrInfo(ip_address,port) == -1)
                {
                    log_error("GetIPAddrInfo failed(err=%d)",CSocketWrapper::GetSockError());
                    return -1;
                }
                uint16_t local_port = 0;
                if (m_socket.BindRandPort(local_port) == -1)
                {
                    log_warning("proxy bind failed");
                    if (m_proxy.GetProxyType() == kProxySocksVer5)
                    {
                        m_socket.Close();
                        return -1;
                    }
                }
                if (m_proxy.GetProxyType() == kProxySocksVer5)
                {
                    std::string local_ipaddress;
                    if (m_socket.GetFirstIPAddress(local_ipaddress) == -1)
                    {
                        log_warning("can not get local ip address");
                        m_socket.Close();
                        return -1;
                    }
                    if (m_udp_associate->Connect(local_ipaddress.c_str(),local_port,IPPROTO_TCP) == -1)
                    {
                        log_warning("udp associate conect failed");
                        m_socket.Close();
                        return -1;
                    }

                    std::string proxy_reply;
                    uint16_t proxy_reply_port = 0;
                    if (m_proxy.UdpAssociate(m_udp_associate->GetSocket(),
                        ip_address,local_port,proxy_reply,proxy_reply_port) == -1)
                    {
                        log_warning("proxy.udp associate failed");
                        m_udp_associate->Disconnect();
                        m_socket.Close();
                        return -1;
                    }
                    m_ipaddress	= proxy_reply;
                    m_port		= proxy_reply_port;
                    if (m_socket.GetIPAddrInfo(m_ipaddress.c_str(),m_port) == -1)
                    {
                        log_error("GetAddrInfo proxy failed(err=%d)",CSocketWrapper::GetSockError());
                        m_udp_associate->Disconnect();
                        m_socket.Close();
                        return -1;
                    }
                }
            }

            m_send_buffer.reset();
            m_recv_buffer.reset();

            ++m_exit_flag_recv;
            uint32_t threadId = 0;
            m_thread_recv = ThreadWrapper::CreateThread(ThreadFunRecv,this,kNormalPriority,"socket recv thread");
            m_thread_recv->Start(threadId);
        }
        CriticalSectionScoped cs(&m_critSect);
        if (m_observer)
        {
            m_observer->OnConnected();
        }

        return 0;
    }

    int32_t CSocketImpl::Disconnect()
    {
        //CriticalSectionScoped cs(&m_crit_interface);
        while (m_exit_flag_recv.Value() > 0)
        {
            --m_exit_flag_recv;
        }
        while (m_exit_flag_send.Value() > 0)
        {
            --m_exit_flag_send;
        }
        m_event_send.Set();

        m_socket.Close();

        StopThread(&m_thread_send);
        StopThread(&m_thread_recv);

        return 0;
    }

    void CSocketImpl::StopThread(ThreadWrapper** thread_)
    {
        ThreadWrapper* tmpPtr = *thread_;
        *thread_ = NULL;
        if (tmpPtr)
        {
            tmpPtr->SetNotAlive();

            if (tmpPtr->Stop())
            {
                delete tmpPtr;
            }
            else
            {
                log_error("stop thread failed");
            //    assert(0);
            }
        }
    }

    int32_t CSocketImpl::Send(const char* buff,uint32_t bytes)
    {
        if (m_socket_type == IPPROTO_TCP)
        {
            m_send_buffer.put(buff, bytes);
            m_event_send.Set();

            return bytes;
        }
        else
        {
            int32_t numBytes = 0;
            if (m_proxy.GetProxyType() == kProxySocksVer5)
            {
                udpheader header = m_proxy.GetUdpHeader(m_ipaddress.c_str(),m_port);
                m_send_buffer.put((const char*)&header,sizeof(udpheader));
                m_send_buffer.put(buff,bytes);
                numBytes = m_socket.SendTo(m_send_buffer.rd_buf(), m_send_buffer.rd_size());
                if (numBytes != (int32_t)m_send_buffer.rd_size())
                {
                    log_error("udp proxy send failed(err=%d)",CSocketWrapper::GetSockError());
                    m_send_buffer.rd_flip(m_send_buffer.rd_size());

                    return -1;
                }
                m_send_buffer.rd_flip(m_send_buffer.rd_size());
            }
            else
            {
                numBytes = m_socket.SendTo(buff, bytes);
                if (numBytes != (int32_t)bytes)
                {
                    log_error("udp send failed(err = % d)",CSocketWrapper::GetSockError());

                    return -1;
                }
            }
            return numBytes;
        }
    }

    void CSocketImpl::SetProxy(
        ProxyType proxy,
        const char* host,
        uint16_t port,
        const char* username,
        const char* password,
        const char* domain)
    {
        m_proxy.SetProxy(proxy,host,port,username,password,domain);

        if (m_udp_associate)
        {
            m_udp_associate->SetProxy(proxy, host, port, username, password, domain);
        }
    }

    bool CSocketImpl::ThreadFunSend(void* args)
    {
        return static_cast<CSocketImpl*>(args)->ThreadProcessSend(args);
    }

    bool CSocketImpl::ThreadProcessSend(void* args)
    {
        log_info("enter send thread");
        BufferQueue send_buffer;
        while (m_exit_flag_send.Value() > 0)
        {
            if (send_buffer.rd_size() == 0)
            {
                if (m_send_buffer.rd_size() == 0)
                {
                    m_event_send.Wait(0xFFFFFFFF);
                    if (m_exit_flag_send.Value() == 0)
                    {
                        break;
                    }
                    if (m_send_buffer.rd_size() == 0)
                    {
                        continue;
                    }
                }
                send_buffer.put(m_send_buffer.rd_buf(),m_send_buffer.rd_size());
                m_send_buffer.rd_flip(m_send_buffer.rd_size());
            }

            int32_t ret = m_socket.Send(send_buffer.rd_buf(),send_buffer.rd_size());
            if (ret <= 0)
            {
                log_error("send failed(err=%d)",CSocketWrapper::GetSockError());
                CriticalSectionScoped cs(&m_critSect);
                if (m_observer)
                {
                    m_observer->OnSendFailed();
                }
                break;
            }
            else
            {
                send_buffer.rd_flip(uint32_t(ret));
            }
        }
        log_info("leave send thread");
        return false;
    }
    bool CSocketImpl::ThreadFunRecv(void* args)
    {
        return static_cast<CSocketImpl*>(args)->ThreadProcessRecv(args);
    }

    bool CSocketImpl::ThreadProcessRecv(void* args)
    {
        log_info("enter recv thread");
        const int32_t buffer_length = 1024 * 10;
        char buff[buffer_length];
        if (m_socket_type == IPPROTO_TCP)
        {
            while (m_exit_flag_recv.Value() > 0)
            {
                int32_t ret = m_socket.Select();
                if (ret < 0)
                {
                    if (m_exit_flag_recv.Value() == 0)
                    {
                        break;
                    }
                    log_error("socket select failed(ret=%d,err=%d)",ret,CSocketWrapper::GetSockError());
                    CriticalSectionScoped cs(&m_critSect);
                    if (m_observer)
                    {
                        m_observer->OnRecvFailed();
                    }
                    break;
                }
                else if (ret > 0)
                {
                    int32_t recv_size = m_socket.Recv(buff, buffer_length);
                    if (recv_size == 0)
                    {
                        SleepMs(5);
                        continue;
                    }
                    if (recv_size <= 0)
                    {
                        if (m_exit_flag_recv.Value() > 0) // ?????????????
                        {
                            log_error("socket recv failed(recv_size=%d,err=%d)",recv_size,CSocketWrapper::GetSockError());
                            CriticalSectionScoped cs(&m_critSect);
                            if (m_observer)
                            {
                                m_observer->OnRecvFailed();
                            }
                        }
                        break;
                    }

                    m_recv_buffer.put(buff,recv_size);

                    while (m_recv_buffer.rd_size() > 0)
                    {
                        if (m_recv_buffer.rd_size() <= sizeof(int32_t))
                        {
                            break;
                        }

                        char* pRecvData = m_recv_buffer.rd_buf();

                        int32_t magic = 0;
                        memcpy((char*)&magic, pRecvData, sizeof(int32_t));
                        magic = ntohl(magic);

                        if (magic != Y_RD_MSG_MAGIC)
                        {
                            m_recv_buffer.rd_flip(recv_size);
                            break;
                        }

                        int32_t msglen = 0;
                        memcpy((char*)&msglen, &pRecvData[4], sizeof(int32_t));

                        if (ntohl(msglen) < sizeof(xMsgHeadReq_t))
                        {
                            m_recv_buffer.rd_flip(recv_size);
                            break;
                        }

                        uint32_t buf_len = ntohl(msglen);
                        if (buf_len > Y_RD_SERVICE_MESSAGE_BLOCK_LEN)
                        {
                            m_recv_buffer.rd_flip(recv_size);
                            break;
                        }

                        if (m_recv_buffer.rd_size() < (uint32_t)buf_len + 8)
                        {
                            break;
                        }
                        {
                            CriticalSectionScoped cs(&m_critSect);

                            char* data = new char[buf_len];
                        //    autoptr_arr<char> arrdata(data);
                            memcpy(data, &pRecvData[8], buf_len);

                            m_observer->OnRecvMessage(std::make_shared<ProcessMessage>(data, buf_len));
                        }
                        m_recv_buffer.rd_flip(buf_len+8);
                    }
                }
                else if (ret == 0)
                {
                    if (m_exit_flag_recv.Value() == 0)
                    {
                        break;
                    }
                    // select??????????????????????????????????????
                    log_error("socket select failed(ret=%d,err=%d)",ret,CSocketWrapper::GetSockError());
                    assert(0);
                }
            }
        }
        else
        {
            sockaddr_in props;
            int32_t proxyOffset = 0;
            while (m_exit_flag_recv.Value() > 0)
            {
                int32_t ret = m_socket.Select();
                if (m_exit_flag_recv.Value() == 0)
                {
                    break;
                }
                if (ret < 0)
                {
                    log_error("udp socket select failed(ret=%d,err=%d)",ret,CSocketWrapper::GetSockError());
                    CriticalSectionScoped cs(&m_critSect);
                    if (m_observer)
                    {
                        m_observer->OnRecvFailed();
                    }
                    break;
                }
                else if (ret > 0)
                {
                    int32_t numBytes = m_socket.Recvfrom(buff, buffer_length,&props);
                    if (numBytes < 0)
                    {
                        if (m_exit_flag_recv.Value() == 0)
                        {
                            break;
                        }
                        log_warning("error while reading(err=%d)",CSocketWrapper::GetSockError());
                        SleepMs(200);
                        continue;
                    }
                    if (numBytes < 3)
                    {
                        log_error("packet is not big enough");
                        continue;
                    }
                    if (m_proxy.GetProxyType() == kProxySocksVer5)
                    {
                        if (numBytes < (int32_t)sizeof(udpheader))
                        {
                            log_error("proxy error while reading,read size error");
                            continue;
                        }
                        proxyOffset = sizeof(udpheader);
                    }
                    else
                    {
                        proxyOffset = 0;
                    }

                    if (!m_socket.CompareAddrInfo(&props))
                    {
                        log_error("got a packet that is not from the server");
                        continue;
                    }
                    CriticalSectionScoped cs(&m_critSect);
                    if (m_observer)
                    {
                        m_observer->OnRecvMessage2(buff + proxyOffset,numBytes - proxyOffset);
                    }
                }
                else if (ret == 0)
                {
                    // select??????????????????????????????????????
                    log_error("socket select failed(ret=%d,err=%d)", ret, CSocketWrapper::GetSockError());
                    assert(0);
                    SleepMs(200);
                }
            }
        }

        log_info("leave recv thread");
        return false;
    }
}
